package com.utopa.canal.client.config;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.utopa.canal.client.annotation.CanalListener;
import com.utopa.canal.client.handler.CanalMessageListenerProxy;
import com.utopa.canal.client.handler.DefaultRowDataHandler;
import com.utopa.canal.client.listener.CanalMessageListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;

import java.lang.reflect.Method;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * @author authorZhao
 * @date 2020年04月15日
 */
@Slf4j
@Configuration
@EnableConfigurationProperties({CanalProperties.class})
@Import({CanalFactory.class, CanalApplicationEvent.class, DefaultRowDataHandler.class})
@ConditionalOnProperty(value = CanalProperties.CANAL_PREFIX_ENABLED, havingValue = "true", matchIfMissing = false)
public class CanalAutoConfiguration implements BeanPostProcessor, ApplicationListener<CanalApplicationEvent> {
    @Autowired
    private CanalConnector canalConnector;
    @Autowired
    private CanalProperties canalProperties;
    private Map<String, CanalMessageListener> canalMessageListenerMap = new ConcurrentHashMap<>();

    @Override
    public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
        return bean;
    }

    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        Class<?> targetClass = bean.getClass();
        // 获取当前类中的所有方法
        Method[] classMethodList = targetClass.getMethods();
        // 筛选出被 @CanalListener 注解的方法
        List<Method> canalListenerMethodList = Arrays.stream(classMethodList).filter(method -> method.isAnnotationPresent(CanalListener.class)).collect(Collectors.toList());
        if (CollectionUtils.isNotEmpty(canalListenerMethodList)) {
            canalListenerMethodList.forEach(method -> {
                // 获取方法上的 CanalListener 注解
                CanalListener canalListener = method.getAnnotation(CanalListener.class);
                // 注解的属性是否为空
                if (StringUtils.isEmpty(canalListener.databaseName()) || StringUtils.isEmpty(canalListener.tableName())) {
                    return;
                }
                // 获取方法的所有参数类型
                Class[] parameterTypes = method.getParameterTypes();
                if (ArrayUtils.isNotEmpty(parameterTypes) && parameterTypes.length >= 2) {
                    // 方法的第二个参数必须是 CanalEntry.EventType 类型
                    if (CanalEntry.EventType.class == parameterTypes[1]) {
                        log.info("添加 CanalListener 监听器定义，class={}，method={}", bean.getClass().getName(), method.getName());
                        CanalMessageListener canalMessageListener = CanalMessageListenerProxy.createProxyObject(bean, method, canalListener);
                        canalMessageListenerMap.put(canalListener.databaseName() + canalListener.tableName(), canalMessageListener);
                    }
                }
            });
            log.info("正在为类 {} 注册 CanalListener 监听器，需要注册的数量为{}", bean.getClass().getName(), canalListenerMethodList.size());
        }
        return bean;
    }

    @Override
    public void onApplicationEvent(CanalApplicationEvent event) {
        log.info("canal 客户端开始处理数据");
        new Thread(() -> {
            int failCount = 0;
            while (true) {
                int batchSize = canalProperties.getBatchSize();
                // 获取指定数量的数据
                Message message = canalConnector.getWithoutAck(batchSize);
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    try {
                        Thread.sleep(canalProperties.getSleepTime());
                    } catch (InterruptedException e) {
                        log.error("当前线程{}执行 sleep() 异常", Thread.currentThread().getName(), e);
                    }
                } else {
                    try {
                        List<CanalEntry.Entry> entries = message.getEntries();
                        for (CanalEntry.Entry entry : entries) {
                            String key = entry.getHeader().getSchemaName() + entry.getHeader().getTableName();
                            CanalMessageListener listener = canalMessageListenerMap.get(key);
                            if (listener != null) {
                                listener.onMessage(entry);
                            }
                        }
                    } catch (Exception e) {
                        log.error("消费 canal 数据异常，batchId={}", batchId, e);
                        if (failCount <= 3) {
                            // 消费失败，回滚
                            failCount++;
                            canalConnector.rollback();
                        } else {
                            // 消费失败次数过多，停止消费，停止回滚
                            failCount = 0;
                        }
                    }
                }
                // 提交确认
                canalConnector.ack(batchId);
            }
        }).start();
    }
}
